
from datetime import datetime
from kafka import KafkaConsumer, KafkaProducer
import time
import json
import queue


class KafkaServer:
    def __init__(self, host, time_type):
        self.host = host
        self.time_type = time_type  # "%Y-%m-%dT%H:%M:%S.%f"
        self.data_queue_dict = {}

    def send_file_data(self, topic, log_path):
        # kafka按行发送离线文件的方法

        producer = KafkaProducer(bootstrap_servers=self.host, key_serializer=str.encode,
                                 value_serializer=lambda x: json.dumps(x).encode('utf-8'))
        with open(log_path, 'r', encoding='utf-8') as f:
            current_time = None
            while True:
                message = f.readline()
                if not message:  # 如果文件读取完毕，则退出循环
                    break
                try:
                    msg = json.loads(message.strip())
                    # 离线数据模拟发送在线数据
                    producer.send(topic, value=msg, key="key")
                    # print("发送数据", msg)
                    t = datetime.strptime(msg["timestamp"], self.time_type)
                    if current_time is not None:
                        sleep_duration = (t - current_time).total_seconds()
                        if sleep_duration > 0:
                            # print("当前时间:", t, " 跳过:", sleep_duration)
                            time.sleep(sleep_duration)
                            # time.sleep(0.1)
                    current_time = t

                except json.JSONDecodeError:
                    print("忽略非JSON格式的日志行")
                    continue

    def send_data(self, topic, data):
        # kafka发送data数据的方法

        # 发送数据
        producer = KafkaProducer(bootstrap_servers=self.host, key_serializer=str.encode,
                                 value_serializer=lambda x: json.dumps(x).encode('utf-8'))
        while True:
            data["time"] = time.time()
            # 事件检测结果
            producer.send(topic, value=data, key="key")
            print("发送数据", data)
            time.sleep(0.1 / 10)

    def consumer_data(self, topic):
        print("接收卡口车辆的实时数据", 'kafka 消费被触发')
        # 首先定义全局变量，用于在不同方法内互相通信
        consumer = KafkaConsumer(
            topic,
            bootstrap_servers=self.host,
            auto_offset_reset='latest',
            enable_auto_commit=False
        )
        self.data_queue_dict[topic] = queue.Queue(maxsize=2000)
        # 将消费者的偏移量设置为最后一条消息的偏移量
        while True:
            time.sleep(0.1 / 20)
            for message in consumer:
                if message:
                    try:
                        message_value = message.value.decode('utf-8')
                        # 根据发送数据不同的数据格式选择
                        vehicle_info = json.loads(message_value)
                        # TODO 接收到的数据，这里添加处理逻辑
                        # print("接收数据", vehicle_info)
                        if self.data_queue_dict[topic].qsize() < 2000:
                            self.data_queue_dict[topic].put(vehicle_info)
                    except Exception as e:
                        print(str(e))

    def get_messages(self, topic):
        return self.data_queue_dict[topic]


if __name__ == "__main__":
    kafka_server = KafkaServer("106.120.201.126:14576", "%Y-%m-%dT%H:%M:%S.%f")
    data = {"a": 111}
    # kafka_server.send_data("SEND_EVENTS", data)
    producer = KafkaProducer(bootstrap_servers="106.120.201.126:14576", key_serializer=str.encode,
                             value_serializer=lambda x: json.dumps(x).encode('utf-8'))
    while True:
        data["time"] = time.time()
        # 事件检测结果
        producer.send("SEND_EVENTS", value=data, key="key")
        print("发送数据", data)
        time.sleep(0.1 / 10)